package com.epic.engine.streams;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;

import java.io.Serializable;
import java.util.Properties;

public class StreamSink implements Serializable {
    private String kafkaHost;
    private String topic;
    public StreamSink(String topic, String kafkaHost){
        this.kafkaHost = kafkaHost;
        this.topic = topic;
    }
    public KafkaSink<String> initSink(){
        Properties properties = new Properties();
        properties.setProperty("client.id","producer-engine");
        properties.setProperty("bootstrap.servers", this.kafkaHost);

        KafkaSink<String> sink = KafkaSink.<String>builder()
                .setBootstrapServers(kafkaHost)
                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                        .setTopic(topic)
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build()
                ).build();
        return sink;
    }
}
